package com.atguigu.flinksql;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Flink05_SQL_BaseUse {
    public static void main(String[] args) {
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port",10000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        env.setParallelism(2);

        //创建表环境
        DataStreamSource<WaterSensor> waterSensorStream =
                env.fromElements(new WaterSensor("sensor_1", 1000L, 10),
                        new WaterSensor("sensor_1", 2000L, 20),
                        new WaterSensor("sensor_2", 3000L, 30),
                        new WaterSensor("sensor_1", 4000L, 40),
                        new WaterSensor("sensor_1", 5000L, 50),
                        new WaterSensor("sensor_2", 6000L, 60));


        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        Table table = tEnv.fromDataStream(waterSensorStream);

        //1、查询一个未注册的表
        Table result = tEnv.sqlQuery(" select * from " + table + " where id='sensor_1'");
        result.printSchema();
        //数据输出
        result.execute().print();

        //2、查询注册过的表
        tEnv.createTemporaryView("sensor",table);


        Table result1 = tEnv.sqlQuery(" select * from sensor ");
        System.out.println("--------------------");
        System.out.println("对表进行注册过后的元数据！");
        result1.printSchema();
        result1.execute().print();



    }
}
